package org.example;

import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * Flink Table API & SQL 使用示例
 * 本示例展示如何使用 Flink 的 Table API 和 SQL 进行流处理：
 * 1. 从文本流中读取数据，转换为 Table。
 * 2. 使用 SQL 查询统计单词出现次数。
 * 3. 将结果转换回 DataStream 并输出。
 * 适合初学者学习 Flink Table API 和 SQL 的基本使用。
 */
public class TableAPISQLExample {

    public static void main(String[] args) throws Exception {
        // 创建流处理执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 设置并行度为 1，方便观察输出
        env.setParallelism(1);

        // 创建 Table 环境
        EnvironmentSettings settings = EnvironmentSettings.newInstance().inStreamingMode().build();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, settings);

        // 模拟输入数据流（实际应用中可能来自 Kafka、Socket 等）
        DataStream<String> textStream = env.fromElements(
            "Hello Flink",
            "Hello World",
            "Hello Flink"
        );

        // 将 DataStream 转换为 Table
        Table inputTable = tableEnv.fromDataStream(textStream, "word");

        // 使用 SQL 查询统计单词出现次数
        String sql = "SELECT word, COUNT(*) as count FROM " + inputTable + " GROUP BY word";
        Table resultTable = tableEnv.sqlQuery(sql);

        // 将结果转换回 DataStream 并输出
        DataStream<org.apache.flink.types.Row> resultStream = tableEnv.toDataStream(resultTable);
        resultStream.print("Table API & SQL 处理结果");

        // 执行流处理作业
        env.execute("Table API & SQL Example");
    }
} 